package com.partsoft.dits.mq;

import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Executor;

import org.apache.commons.pool.KeyedObjectPool;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import com.ibm.mq.MQC;
import com.ibm.mq.MQException;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.partsoft.dits.AbstractConfigurableDescriptorManager;
import com.partsoft.dits.AbstractMessageByteBatchProcessor;
import com.partsoft.dits.DataBroadcaster;
import com.partsoft.dits.DataEdgeDescriptor;
import com.partsoft.dits.DataIntegrateBatch;
import com.partsoft.dits.DescriptorAccepter;
import com.partsoft.dits.DescriptorDepends;
import com.partsoft.dits.DescriptorManager;
import com.partsoft.dits.DescriptorRegistry;
import com.partsoft.dits.EdgeDescriptor;

@SuppressWarnings("unchecked")
public class ContextConfigurableMQQueueBroadcastReactor extends
		AbstractConfigurableDescriptorManager<MQQueueBroadcastDescriptor>
		implements DescriptorRegistry<MQQueueBroadcastDescriptor>,
		DescriptorManager<MQQueueBroadcastDescriptor>,
		DataBroadcaster,
		ApplicationListener<ApplicationEvent>,
		ApplicationContextAware, 
		DescriptorDepends<MQQueueBroadcastDescriptor> {
	
	private MQPoolProvider mqPoolProvider;
	
	private Executor taskExecutor;
	
	private Set<Class<? extends EdgeDescriptor>> dependsClasses;
	{
		dependsClasses = new HashSet<Class<? extends EdgeDescriptor>>(1);
		dependsClasses.add(MQPoolDescriptor.class);
	}
	
	private Map<Class<? extends DataEdgeDescriptor>, Map<String, Set<MQQueueBroadcastDescriptor>>> dataClassRelationMap = Collections.synchronizedMap(new HashMap<Class<? extends DataEdgeDescriptor>, Map<String, Set<MQQueueBroadcastDescriptor>>>(5));
	
	private Map<String, MQConnectPoolRef> mqMessageTransmitters = Collections.synchronizedMap(new HashMap<String, MQConnectPoolRef>(2));
	
	private AbstractMessageByteBatchProcessor byteProcessor = new AbstractMessageByteBatchProcessor() {
		@Override
		protected Object doProcess(DataEdgeDescriptor dataDescriptor, byte[] bytes, DescriptorAccepter<EdgeDescriptor> accepter) {
			doBroadcast((Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(dataDescriptor), dataDescriptor, bytes, accepter);
			return null;
		}
		
		protected void validateBatch(DataIntegrateBatch batch) {
			super.validateBatch(batch);
			validateSupportDataDescriptor(batch.getDataDescriptor());
		};
	};
	
	public ContextConfigurableMQQueueBroadcastReactor() {}
	
	public void setMQPoolProvider(MQPoolProvider mqPoolProvider) {
		assertNoInit();
		this.mqPoolProvider = mqPoolProvider;
	}
	
	public MQPoolProvider getMQPoolProvider() {
		return mqPoolProvider;
	}
	
	public Executor getTaskExecutor() {
		return taskExecutor;
	}

	public void setTaskExecutor(Executor taskExecutor) {
		this.taskExecutor = taskExecutor;
	}

	public Class<MQQueueBroadcastDescriptor> getSupportClass() {
		return MQQueueBroadcastDescriptor.class;
	}

	@Override
	protected void beforeRemoveDescriptor(MQQueueBroadcastDescriptor descriptor) {
		removeDataDescriptorFromMap(descriptor);
	}

	@Override
	protected void postRemoveDescriptor(MQQueueBroadcastDescriptor descriptor) {
		
	}

	@Override
	protected void beforeLoadConfiguredDescriptor(MQQueueBroadcastDescriptor descriptor) {
		addDataDescriptorToMap(descriptor);
	}
	
	@Override
	protected void beforeUnLoadConfiguredDescriptor(MQQueueBroadcastDescriptor descriptor) {
		removeDataDescriptorFromMap(descriptor);
	}
	
	private void removeDataDescriptorFromMap(MQQueueBroadcastDescriptor descriptor) {
		if (dataClassRelationMap.containsKey(descriptor.getDataClass())) {
			Map<String, Set<MQQueueBroadcastDescriptor>> clazzIds =  dataClassRelationMap.get(descriptor.getDataClass());
			if (clazzIds.containsKey(descriptor.getDataDescriptorId())) {
				Set<MQQueueBroadcastDescriptor> qes = clazzIds.get(descriptor.getDataDescriptorId());
				qes.remove(descriptor);
				if (qes.size() == 0) {
					clazzIds.remove(descriptor.getDataDescriptorId());
					if (clazzIds.size() == 0) {
						dependsClasses.remove(descriptor.getDataClass());
						dataClassRelationMap.remove(descriptor.getDataClass());
					}
				}
			}
		}
		
		if (mqMessageTransmitters.containsKey(descriptor.getConnectPool())) {
			MQConnectPoolRef ref = mqMessageTransmitters.get(descriptor.getConnectPool());
			if (ref.removeRef() == 0) {
				mqMessageTransmitters.remove(descriptor.getConnectPool());
			}
		}
	}
	
	private void addDataDescriptorToMap(MQQueueBroadcastDescriptor descriptor) {
		Map<String, Set<MQQueueBroadcastDescriptor>> clazzIds =  dataClassRelationMap.get(descriptor.getDataClass());
		if (clazzIds == null) { 
			clazzIds = Collections.synchronizedMap(new HashMap<String, Set<MQQueueBroadcastDescriptor>>(5));
			dataClassRelationMap.put(descriptor.getDataClass(), clazzIds);
			dependsClasses.add(descriptor.getDataClass());
		}
		Set<MQQueueBroadcastDescriptor> descs = clazzIds.get(descriptor.getDataDescriptorId());
		if (descs == null) {
			descs = Collections.synchronizedSet(new HashSet<MQQueueBroadcastDescriptor>(2));
			clazzIds.put(descriptor.getDataDescriptorId(), descs);
		}
		descs.add(descriptor);
		
		MQConnectPoolRef transmitter = null;
		if (mqMessageTransmitters.containsKey(descriptor.getConnectPool())) {
			transmitter = mqMessageTransmitters.get(descriptor.getConnectPool());
		} else {
			KeyedObjectPool pool = getMQPoolProvider().getMQPool(descriptor.getConnectPool());
			MQMessageTransmitter mqt = new MQMessageTransmitter();
			mqt.setQmPool(pool);
			mqt.setTaskExecutor(getTaskExecutor());
			transmitter = new MQConnectPoolRef(mqt);
			mqMessageTransmitters.put(descriptor.getConnectPool(), transmitter);
		}
		transmitter.addRef();
	}
	
	@Override
	protected void setupDescriptorIdentifier(MQQueueBroadcastDescriptor descriptor, String id) {
		descriptor.setId(id);
	}

	@Override
	protected Properties beforeAddDescriptor(MQQueueBroadcastDescriptor descriptor) {
		Properties props = new Properties();
		props.setProperty("connectPool", descriptor.getConnectPool());
		props.setProperty("queueManager", descriptor.getQueueManager());
		props.setProperty("queue", descriptor.getQueue());
		props.setProperty("dataClass", descriptor.getDataClass().getName());
		props.setProperty("dataDescriptorId", descriptor.getDataDescriptorId());
		props.setProperty("title", descriptor.getTitle());
		props.setProperty("description", descriptor.getDescription());
		props.setProperty("ordinal", Integer.valueOf(descriptor.getOrdinal()).toString());
		return props;
	}
	
	@Override
	protected void postAddDescriptor(
			MQQueueBroadcastDescriptor descriptor) {
		addDataDescriptorToMap(descriptor);
	}

	@Override
	protected void validateDescriptor(MQQueueBroadcastDescriptor descriptor) {
		Assert.notNull(descriptor.getDataClass(), "必须设置数据类。");
		Assert.hasText(descriptor.getDataDescriptorId(), "必须设置数据描述符ID。");
		Assert.hasText(descriptor.getConnectPool(), "必须设置MQ池。");
		Assert.hasText(descriptor.getQueueManager(), "必须设置MQ管理器。");
		Assert.hasText(descriptor.getQueue(), "必须设置MQ发送队列。");
		
		//校验MQ池
		Assert.isTrue(getMQPoolProvider().hasMQPool(descriptor.getConnectPool()), String.format("找不到指定的MQ池(%s)。", descriptor.getConnectPool()));
		KeyedObjectPool qmPool = getMQPoolProvider().getMQPool(descriptor.getConnectPool());
		Object qmObject = null;
		try {
			qmObject = qmPool.borrowObject(descriptor.getQueueManager());
		} catch (Throwable e) {
			throw new IllegalArgumentException(String.format("MQ池(%s)中找不到队列管理器(%s)", descriptor.getConnectPool(), descriptor.getQueueManager()), e);
		}
		MQQueueManager qmgr = (MQQueueManager) qmObject;
		MQQueue queue = null;
		try {
			queue = qmgr.accessQueue(descriptor.getQueue(), MQC.MQOO_INPUT_SHARED | MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING);
		} catch (MQException e) {
			throw new IllegalArgumentException(String.format("打开MQ池(%s)队列管理器(%s)中的队列(%s)出错。", descriptor.getConnectPool(), descriptor.getQueueManager(), descriptor.getQueue()), e);
		} finally {
			if (queue != null) {
				try {
					queue.close();
				} catch (MQException e) {}
			}
			try {
				qmPool.returnObject(descriptor.getQueueManager(), qmgr);
			} catch (Exception e) {
				throw new IllegalArgumentException(e.getMessage(), e);
			}
		}
	}

	@Override
	protected String resolveDescriptorIdentifier(MQQueueBroadcastDescriptor descriptor) {
		if (StringUtils.hasText(descriptor.getId())) return descriptor.getId();
		StringBuffer buffer = new StringBuffer();
		buffer.append(descriptor.getConnectPool()).append(".").
			append(descriptor.getQueueManager()).append(".").
			append(descriptor.getQueue()).append(".").
			append(descriptor.getDataClass() != null ? descriptor.getDataClass().getName() : "nodata").append(".").
			append(descriptor.getDataDescriptorId());
		return buffer.toString();
	}
	
	@Override
	public void initContext() {
		Assert.notNull(getMQPoolProvider(), "必须设置MQ池。");
		super.initContext();
	}
	
	protected Class<? extends DataEdgeDescriptor> validateSupportDataDescriptor(DataEdgeDescriptor dataDescriptor) {
		Class<? extends DataEdgeDescriptor> descriptorClazz = (Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(dataDescriptor);
		Assert.isTrue(supportData(dataDescriptor), String.format("不支持此数据(%s)，描述符(%s)。", descriptorClazz.getName(), dataDescriptor.getId()));
		return descriptorClazz;
	}
	
	//implements DataBroadcaster
	public void broadcast(DataEdgeDescriptor dataDescriptor,  byte[] bytes) {
		broadcast(dataDescriptor, bytes, null);
	}
	
	public void broadcast(DataEdgeDescriptor dataDescriptor, byte[] bytes, DescriptorAccepter<EdgeDescriptor> accepter) {
		assertAlreadyInit();
		Assert.notNull(dataDescriptor);
		Assert.notNull(bytes);
		Class<? extends DataEdgeDescriptor> descriptorClazz = validateSupportDataDescriptor(dataDescriptor);
		doBroadcast(descriptorClazz, dataDescriptor, bytes, accepter);
	}

	private void doBroadcast(Class<? extends DataEdgeDescriptor> descriptorClazz, DataEdgeDescriptor dataDescriptor,  byte[] bytes, DescriptorAccepter<EdgeDescriptor> accepter) {
		MessageTransmitter transmitter = null;
		for (MQQueueBroadcastDescriptor mqbid : dataClassRelationMap.get(descriptorClazz).get(dataDescriptor.getId())) {
			if (accepter == null || accepter.accept(mqbid)) {
				transmitter = mqMessageTransmitters.get(mqbid.getConnectPool());
				transmitter.sendMessage(mqbid.getQueueManager(), mqbid.getQueue(), bytes);
			}
		}
	}
	
	public void process(DataIntegrateBatch batch) {
		byteProcessor.process(batch);
	}

	public boolean supportData(Class<? extends DataEdgeDescriptor> descriptorClazz) {
		return dataClassRelationMap.containsKey(descriptorClazz);
	}

	public boolean supportData(DataEdgeDescriptor dataDescriptor) {
		Class<? extends DataEdgeDescriptor> clazz = (Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(dataDescriptor);
		return supportData(clazz) ? dataClassRelationMap.get(clazz).containsKey(dataDescriptor.getId()) : false;
	}
	

	@Override
	public Set<Class<? extends EdgeDescriptor>> getDependClasses() {
		return Collections.unmodifiableSet(dependsClasses);
	}

	@Override
	public boolean dependDescriptor(EdgeDescriptor descriptor) {
		boolean result = false;
		Class<?> dClazz = ClassUtils.getUserClass(descriptor);
		if (getDependClasses().contains(dClazz)) {
			if (ClassUtils.isAssignable(MQPoolDescriptor.class, dClazz)) {
				MQPoolDescriptor mqpd = (MQPoolDescriptor) descriptor;
				result = mqMessageTransmitters.containsKey(mqpd.getId());
			} else if (ClassUtils.isAssignable(DataEdgeDescriptor.class, dClazz) && dataClassRelationMap.containsKey(dClazz)) {
				result = dataClassRelationMap.get(dClazz).containsKey(((DataEdgeDescriptor)descriptor).getId());
			}
		}
		return result;
	}


	@Override
	public MQQueueBroadcastDescriptor[] getDescriptorByDepend(EdgeDescriptor descriptor) {
		MQQueueBroadcastDescriptor []returnArray = new MQQueueBroadcastDescriptor[0];
		if (dependDescriptor(descriptor)) {
			Set<MQQueueBroadcastDescriptor> returnList = new LinkedHashSet<MQQueueBroadcastDescriptor>();
			if (ClassUtils.isAssignableValue(MQPoolDescriptor.class, descriptor)) {
				for (MQQueueBroadcastDescriptor e : getDescriptorMap().values()) {
					if (ObjectUtils.nullSafeEquals(e.getConnectPool(), ((MQPoolDescriptor)descriptor).getId())) returnList.add(e);
				}
			} else {
				Map<String, Set<MQQueueBroadcastDescriptor>> rels = dataClassRelationMap.get(ClassUtils.getUserClass(descriptor));
				if (rels != null && rels.containsKey(((DataEdgeDescriptor)descriptor).getId())) {
					returnList.addAll(rels.get(((DataEdgeDescriptor)descriptor).getId()));
				}
			}
			returnArray = returnList.toArray(returnArray);
		}
		return returnArray;
	}


	static class MQConnectPoolRef implements MessageTransmitter {
		
		private final MessageTransmitter delegate;
		
		private int ref = 0;
		
		MQConnectPoolRef(MessageTransmitter delegate) {
			this.delegate = delegate;
		}

		public void sendMessage(String connName, String destName, String msg) {
			delegate.sendMessage(connName, destName, msg);
		}

		public void sendMessage(String connName, String destName, byte[] msg) {
			delegate.sendMessage(connName, destName, msg);
		}
		
		public int addRef() {
			return ++ref;
		}
		
		public int removeRef() {
			return --ref;
		}
		
	}
}
